package org.example.apitest.processfunction;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.example.apitest.beans.SensorReading;

public class ProcessTest3SideOutputCase {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从socket中读取数据
        DataStream<String> inputStream = env.socketTextStream("localhost", 7777);

        // 转换为SensorReading类型
        DataStream<SensorReading> dataStream = inputStream
                .map((MapFunction<String, SensorReading>) SensorReading::new);

        // 定义一个OutputTag,用来表示侧输出流,也就是低温流
        OutputTag<SensorReading> lowTempTag = new OutputTag<SensorReading>("low-temp") {
        };

        // 测试ProcessFunction,自定义测输出流实现分流操作
        SingleOutputStreamOperator<Object> highTempStream = dataStream.process(new ProcessFunction<SensorReading, Object>() {

            @Override
            public void processElement(SensorReading value, ProcessFunction<SensorReading, Object>.Context ctx, Collector<Object> out) throws Exception {
                // 判断温度,大于30是高温流,输出到主流,反之输出到低温流
                if (value.getTemperature() > 30) {
                    out.collect(value);
                } else {
                    ctx.output(lowTempTag, value);
                }
            }
        });
        highTempStream.print("high");
        highTempStream.getSideOutput(lowTempTag).print("low");

        env.execute();
    }
}
